package org.example.wordcount;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * world count(dataStream)
 */
public class WordCountStream {
    public static void main(String[] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        创建带有webui的flink运行环境，需要flink-runtime-web依赖，本地开发使用 http://127.0.0.1:8081
//        在没有指定并行度时，并行度默认为电脑物理线程数
        StreamExecutionEnvironment env = StreamExecutionEnvironment
                .createLocalEnvironmentWithWebUI(new Configuration());
//        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
//        全局设置并行度
        env.setParallelism(3);
        DataStreamSource<String> aa = env.readTextFile("aa");
//        SingleOutputStreamOperator<Tuple2<String, Integer>> ds = aa.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
//            @Override
//            public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
//                String[] split = s.split(" ");
//                for (String str : split) {
//                    Tuple2<String, Integer> tuple2 = Tuple2.of(str, 1);
//                    collector.collect(tuple2);
//                }
//            }
//        });
        SingleOutputStreamOperator<Tuple2<String, Integer>> ds = aa.flatMap(
                        (String s, Collector<Tuple2<String, Integer>> collector) -> {
                            String[] split = s.split(" ");
                            for (String str : split) {
                                Tuple2<String, Integer> tuple2 = Tuple2.of(str, 1);
                                collector.collect(tuple2);
                            }
                        })
//                设置并行度，优先级比全局设置高
                .setParallelism(2)
//                lamda泛型擦除问题，需要指定入参类型
                .returns(Types.TUPLE(Types.STRING, Types.INT));

//        KeyedStream<Tuple2<String, Integer>, String> ks = ds.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
//            @Override
//            public String getKey(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
//                return stringIntegerTuple2.f0;
//            }
//        });
        KeyedStream<Tuple2<String, Integer>, String> ks =
                ds.keyBy((Tuple2<String, Integer> stringIntegerTuple2) -> stringIntegerTuple2.f0);
        SingleOutputStreamOperator<Tuple2<String, Integer>> sum1 = ks.sum(1);
        sum1.print();
        env.execute();

    }
}